package xyz.labradormm.spider.util;

import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author hhj
 * @description
 * @date 2021-03-18 14:22
 */
@Component
@Slf4j
public class RedisDelayedQueue {


    @Autowired
    private RedissonClient redissonClient;

    public abstract static class TaskEventListener<T> {
        /**
         * 执行方法
         *
         * @param t
         */
        public abstract void invoke(T t);
    }


    /**
     * 添加队列
     *
     * @param t        DTO传输类
     * @param delay    时间数量
     * @param timeUnit 时间单位
     * @param <T>      泛型
     */
    @Async
    public <T> void addQueue(T t, long delay, TimeUnit timeUnit) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(t.getClass().getName());
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
        //销毁操作对象，节省内存空间
        delayedQueue.destroy();
    }

    /**
     * 获取队列
     *
     * @param zClass            DTO泛型
     * @param taskEventListener 任务回调监听
     * @param <T>               泛型
     * @return
     */
    public <T> void getQueue(Class zClass, TaskEventListener taskEventListener) {
        RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(zClass.getName());
        //此行代码不能删除，否则会导致服务器关闭时，队列剩余元素不执行，take阻塞，需要等待offer操作。
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        //由于此线程需要常驻，可以新建线程，不用交给线程池管理
        new Thread(() -> {
            while (true) {
                blockingDeque.touch();
                try {
                    T t = blockingDeque.take();
                    taskEventListener.invoke(t);
                } catch (InterruptedException e) {
                    log.error("延时队列获取失败：", e);
                } catch (RedissonShutdownException e) {
                    log.info("延时队列关闭-------------------------------------------------------------------------");
                    break;
                } catch (Exception e) {
                    log.error("延时队列处理失败：", e);
                }
            }
        }).start();
    }


}
